### Motivation
Our organization is currently planning a migration from RocketMQ to Pulsar. To facilitate this transition, we aim to implement a standardized abstraction layer for MQ clients that encapsulates implementation details of specific messaging systems. This abstraction layer will allow seamless engine replacement while maintaining consistent client interfaces. However, two critical compatibility issues hinder the unification of message fetching patterns between RocketMQ/Kafka and Pulsar:

**1. Interface Disparity:**

- Pulsar currently lacks native support for offset-based message fetching (fetch/pull paradigm) that allows specifying exact message positions and batch sizes.

**2. Positioning Mechanism Mismatch:**

- RocketMQ/Kafka: Utilize monotonically increasing numerical offsets for message positioning and acknowledgment
- Pulsar: Relies on composite MessageID (ledgerId + entryId + batchIndex) for message identification.

Therefore, I propose to contribute a customized feature implementation to the Pulsar contributor repository that leverages existing Pulsar APIs to achieve functionality similar to RocketMQ's pull consumer. This implementation could serve companies like ours with standardized client requirements, allowing them to reuse this solution.


### Modifications
1. Upgrade Pulsar version to 4.0.8(features include PIP-415 a new API to get the message id by offset) #23 
2. Implement RocketMQ-like pull consumer functionality using existing Pulsar APIs.